MongoDB Spark Connector 实战指南

2023-05-31 16:11| 来源: 网络整理| 查看: 265

Why Spark with MongoDB? 高性能，官方号称 100x faster，因为可以全内存运行，性能提升肯定是很明显的简单易用，支持 Java、Python、Scala、SQL 等多种语言，使得构建分析应用非常简单统一构建，支持多种数据源，通过 Spark RDD 屏蔽底层数据差异，同一个分析应用可运行于不同的数据源；应用场景广泛，能同时支持批处理以及流式处理

MongoDB Spark Connector 为官方推出，用于适配 Spark 操作 MongoDB 数据；本文以 Python 为例，介绍 MongoDB Spark Connector 的使用，帮助你基于 MongoDB 构建第一个分析应用。

准备 MongoDB 环境

安装 MongoDB 参考 Install MongoDB Community Edition on Linux

mkdir mongodata mongod --dbpath mongodata --port 9555 准备 Spark python 环境

参考 PySpark - Quick Guide

下载 Spark cd /home/mongo-spark wget http://mirrors.tuna.tsinghua.edu.cn/apache/spark/spark-2.4.4/spark-2.4.4-bin-hadoop2.7.tgz tar zxvf spark-2.4.4-bin-hadoop2.7.tgz 设置 Spark 环境变量 export SPARK_HOME=/home/mongo-spark/spark-2.4.4-bin-hadoop2.7 export PATH=$PATH:/home/mongo-spark/spark-2.4.4-bin-hadoop2.7/bin export PYTHONPATH=$SPARK_HOME/python:$SPARK_HOME/python/lib/py4j-0.10.4-src.zip:$PYTHONPATH export PATH=$SPARK_HOME/python:$PATH 运行 Spark RDD 示例 # count.py from pyspark import SparkContext sc = SparkContext("local", "count app") words = sc.parallelize ( ["scala", "java", "hadoop", "spark", "akka", "spark vs hadoop", "pyspark", "pyspark and spark"] ) counts = words.count() $SPARK_HOME/bin/spark-submit count.py Number of elements in RDD → 8

如果上述程序运行成功，说明 Spark python 环境准备成功，还可以测试 Spark 的其他 RDD 操作，比如 collector、filter、map、reduce、join 等，更多示例参考 PySpark - Quick Guide

Spark 操作 MongoDB 数据

参考 Spark Connector Python Guide

准备测试数据 test.coll01 插入3条测试数据，test.coll02 未空 mongo --port 9555 > db.coll01.find() { "_id" : 1, "type" : "apple", "qty" : 5 } { "_id" : 2, "type" : "orange", "qty" : 10 } { "_id" : 3, "type" : "banana", "qty" : 15 } > db.coll02.find() 准备操作脚本，将输入集合的数据按条件进行过滤，写到输出集合 # mongo-spark-test.py from pyspark.sql import SparkSession # Create Spark Session spark = SparkSession \ .builder \ .appName("myApp") \ .config("spark.mongodb.input.uri", "mongodb://127.0.0.1:9555/test.coll01") \ .config("spark.mongodb.output.uri", "mongodb://127.0.0.1:9555/test.coll") \ .getOrCreate() # Read from MongoDB df = spark.read.format("mongo").load() df.show() # Filter and Write df.filter(df['qty'] >= 10).write.format("mongo").mode("append").save() # Use SQL # df.createOrReplaceTempView("temp") # some_fruit = spark.sql("SELECT type, qty FROM temp WHERE type LIKE '%e%'") # some_fruit.show() 运行脚本 $SPARK_HOME/bin/spark-submit --packages org.mongodb.spark:mongo-spark-connector_2.11:2.4.1 mongo-spark-test.py mongo --port 9555 > db.coll02.find() { "_id" : 2, "qty" : 10, "type" : "orange" } { "_id" : 3, "qty" : 15, "type" : "banana" }

【本文地址】

公司简介

联系我们